package com.xzx.flink.streamapi.source;

import com.xzx.flink.bean.ClickEvent;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/17 21:47
 */
public class ClickSource  implements ParallelSourceFunction<ClickEvent> {

    private boolean running = true;

    @Override
    public void run(SourceContext<ClickEvent> ctx) throws Exception {
        Random random = new Random();
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1","./prod?id=2"};
        while (running) {
            ctx.collect(new ClickEvent(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], System.currentTimeMillis()));
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
